package com.atguigu.sparkstreaming.demos

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 *    在SparkStreaming中有3个和时间相关的参数
 *        batchDuration: Duration：    将多长时间消费到的数据封装为一个RDD。
 *                                    一个批次的采集时间间隔
 *
 *        window :      Duration :     要计算的数据的时间范围
 *
 *        slide  :      Duration :    多久提交一次Job运算
 *
 *        batchDuration在构造StreamingContext必须指定的。
 *        默认情况，没指定window和slide时，它们都等同于 batchDuration。
 *            今天上午玩的所有案例，都是 消费10s的数据作为1个批次，没间隔10s，提交一个Job，这个Job算过去10s的数据。
 *
 *            slide 和 window 必须是 batchDuration的整倍数。
 *
 *
 *    -------------------------
 *      在获取流的任意位置，调用window()
 *
 *    ---------------------------
 *    	- object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord)
 *          解决方法:
 *                    方法一:不要在获取了初始的ds: InputDStream[ConsumerRecord[String, String]]之后就立刻window
 *                        而应该先取出 ConsumerRecord中的value，之后再window
 *
 *                    方法二：  将 ConsumerRecord使用Kryo 进行序列化。 参考SparkCore 课件
 *
 *
 *     ----------------------
 *        batchDuration只要是 1分钟的因子。
 *        每间隔1分钟(slide)，计算过去2分钟(window)产生的数据！
 *
 *        -----------------
 *            DS[(word,1)].reduceByKeyAndWindow(reduceFunc,window,slide)
 *
 *
 *          DS[(word,1)].reduceByKeyAndWindow(_+_,20s,10s)
 *              等价于
 *          DS[(word,1)].window(20s,10s).reduceByKey(_+_)
 */
object WindowDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "220309",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )



    //指定window和slide   错误示范
    //val ds1: DStream[ConsumerRecord[String, String]] = ds.window(Seconds(10), Seconds(10))

    //(String, Int):(word,n)
    val ds2: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map((_, 1)).reduceByKey(_+_)

    val ds3: DStream[(String, Int)] = ds2.window(Seconds(10), Seconds(15))

    ds3.print(1000)

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
